package com.atguigu.Flink.timeAndwindow;

import com.atguigu.Flink.POJO.Event;
import com.atguigu.Flink.POJO.WordCount;
import com.atguigu.Flink.function.ClickSource;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.TSFBuilder;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.*;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

public class Flink04_WindowAssigner {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<Event> ds = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner(
                                (event, ts) -> event.getTs()
                        )
                );
        ds.print("input");
        //非按键分区
        //时间窗口
//        ds.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(10))).min("ts").print();
//        ds.windowAll(TumblingEventTimeWindows.of(Time.seconds(10))).min("ts").print();
        //计数窗口
//        ds.countWindowAll(5L).min("ts").print("window");
//        ds.countWindowAll(6L,3).min("ts").print("window");

        //按键分区
        //计数窗口
//        ds.map(event -> new WordCount(event.getUser(),1L))
//                .keyBy(
//                k->k.getWord()
//        ).countWindow(4,2).sum("count").print("window");
        //时间窗口
        ds.map(event -> new WordCount(event.getUser(),1L))
                .keyBy(
                        k->k.getWord()
                ).window(
                          //滚动窗口
//                        TumblingProcessingTimeWindows.of(Time.seconds(10)) //处理时间
//                          TumblingEventTimeWindows.of(Time.seconds(10)) //事件时间
                          //滑动窗口
//                        SlidingEventTimeWindows.of(Time.seconds(10),Time.seconds(5))
//                        SlidingProcessingTimeWindows.of(Time.seconds(10),Time.seconds(5))
                         //会话窗口
//                        ProcessingTimeSessionWindows.withGap(Time.seconds(5))
                          EventTimeSessionWindows.withGap(Time.seconds(5))
                ).sum("count").print("window");


        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
